package com.lzwk.app;

import com.lzwk.app.dwd.ocpx.DwdLookUpJoinTables;
import com.lzwk.app.dwd.ocpx.DwdLookUpJoinTablesN;
import com.lzwk.app.dws.ocpx.DwsLookUpJoinTables;
import com.lzwk.app.function.*;
import com.lzwk.app.ods.adb.AdbJDBCTables;
import com.lzwk.app.ods.ocpx.OdsJDBCTables;
import com.lzwk.base.BaseAppV1;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.log4j.Logger;
import org.apache.logging.log4j.Level;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

/**
 * @title: CDC
 * @Author Yeaser
 * @Date: 2022/1/29 15:31
 * @Version 1.0
 * Description: TODO
 */
public class OcpxSum extends BaseAppV1 {

    public static void main(String[] args) {
        new OcpxSum().init(8,"cks","ocpx_tmp","");

    }
    @Override
    protected void run(StreamExecutionEnvironment env,StreamTableEnvironment tenv) {

/*        Logger.getLogger("org.apache.flink").setLevel(Level.getLevel(""));
        Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.INFO);*/

        tenv.getConfig().setIdleStateRetention(Duration.ofHours(3));//未更新的状态保留时间
        Configuration conf = tenv.getConfig().getConfiguration();

        conf.setString("table.exec.mini-batch.enabled","true"); //开启MiniBatch
        conf.setString("table.exec.mini-batch.allow-latency","12s"); //批量输出的间隔时间
        conf.setString("table.exec.mini-batch.size", "80000"); //防止OOM设置每个批次最多缓存数据的条数
        conf.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");//Global两阶段聚合缓解数据倾斜
        conf.setString("table.exec.async-lookup.buffer-capacity", "1000");//异步查找最大I/O数
        conf.setString("table.exec.async-lookup.timeout", "4min");//异步查找超时时间
        conf.setString("sql-client.verbose", "true");//详细error打印
        conf.setString("table.exec.window-agg.buffer-size-limit", "400000");//聚合窗口buffer大小

        tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_new");

        //函数注册
        tenv.createTemporaryFunction("GetJsonString", GetJsonString.class);
        tenv.createTemporaryFunction("GetJsonInt", GetJsonInt.class);
        tenv.createTemporaryFunction("GetJsonDouble", GetJsonDouble.class);
        tenv.createTemporaryFunction("starts_with", StartsWith.class);
        tenv.createTemporaryFunction("GetMixedOrderId", GetMixedOrderId.class);


        String sourceType = "source";
        String sinkType = "sink";
        String insertType = "insert";
        String queryType = "query";


        jdbcTables(tenv);


        ArrayList<String> sourceSqls = ocpxTables().get(sourceType);
        for (String sql : sourceSqls) {

            tenv.executeSql(sql);
        }

        ArrayList<String> sinkSqls = ocpxTables().get(sinkType);
        for (String sql : sinkSqls) {

            tenv.executeSql(sql);
        }

        StatementSet statementSet = tenv.createStatementSet();

        ArrayList<String> insertSqls = ocpxTables().get(insertType);
        for (String insertSql : insertSqls) {
            statementSet.addInsertSql(insertSql);
        }

        try {
            statementSet.execute().wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    private void jdbcTables(StreamTableEnvironment tenv){
        //JDBC表注册
        OdsJDBCTables.ocpxPlatformAccountInfo(tenv);
        OdsJDBCTables.ocpxWxappMpaccountInfo(tenv);
        OdsJDBCTables.ocpxWxappMpadgroupData(tenv);
        OdsJDBCTables.ocpxWxappOceanaccountInfo(tenv);
        OdsJDBCTables.ocpxWxappMpcampaignData(tenv);
        OdsJDBCTables.ocpxWxappOceanadplandata(tenv);
        OdsJDBCTables.ocpxWxappOceancreativeinfo(tenv);
        OdsJDBCTables.ocpxWxappQQauthorizationInfo(tenv);
        OdsJDBCTables.ocpxWxappOceancampaigndata(tenv);
        OdsJDBCTables.ocpxOceanAuthorInfo(tenv);

        OdsJDBCTables.ocpxTagInfo(tenv);
        OdsJDBCTables.ocpxProductProduct(tenv);
        OdsJDBCTables.ocpxProductProductSkin(tenv);
        OdsJDBCTables.ocpxPromotePromotesource(tenv);
        OdsJDBCTables.ocpxPublicAdminRecordorgref(tenv);
        OdsJDBCTables.ocpxWxapp(tenv);
        OdsJDBCTables.ocpxAccountAccount(tenv);
        OdsJDBCTables.ocpxAccountWxuser(tenv);
        OdsJDBCTables.ocpxMpSubFeedBack(tenv);
        OdsJDBCTables.ocpxPayOrder(tenv);

        //ADB注册表
        AdbJDBCTables.adbAccountMapping(tenv);
        AdbJDBCTables.adbDivisionActvity(tenv);
    }

    private Map<String,ArrayList<String>> ocpxTables(){
        //DWS消耗

        String sourceType = "source";
        String sinkType = "sink";
        String insertType = "insert";
        HashMap<String, ArrayList<String>> res = new HashMap<>();

        ArrayList<String> sourceSqlList = new ArrayList<>();
        ArrayList<String> sinkSqlList = new ArrayList<>();
        ArrayList<String> insertSqlList = new ArrayList<>();


/*        //订单CDC+join
        Map<String, String> orderSqls = DwdLookUpJoinTables.dwdPayOrder();
        sourceSqlList.add(orderSqls.get(sourceType));
        sinkSqlList.add(orderSqls.get(sinkType));
        insertSqlList.add(orderSqls.get(insertType));

        //WxuserCDC 和订单表关联拿用户关注（时间前后3年）
        Map<String, String> wxuserSqls = DwdLookUpJoinTables.odsWxuser();
        sourceSqlList.add(wxuserSqls.get(sourceType));*/

/*        //MP&GDT消耗
        Map<String, String> mpSqls = DwsLookUpJoinTables.dwsOcpxWxappMpdailyReportgroupdata();
        sourceSqlList.add(mpSqls.get(sourceType));
        sinkSqlList.add(mpSqls.get(sinkType));
        insertSqlList.add(mpSqls.get(insertType));

        //抖音消耗
        Map<String, String> dySqls = DwsLookUpJoinTables.dwsOcpxWxappOceancreativereportdata();
        sourceSqlList.add(dySqls.get(sourceType));
        sinkSqlList.add(dySqls.get(sinkType));
        insertSqlList.add(dySqls.get(insertType));

        //快手消耗
        Map<String, String> ksSqls = DwsLookUpJoinTables.dwsOcpxKsclUniytreportdata();
        sourceSqlList.add(ksSqls.get(sourceType));
        sinkSqlList.add(ksSqls.get(sinkType));
        insertSqlList.add(ksSqls.get(insertType));

        //抖音性别画像
        Map<String, String> genderSqls = DwsLookUpJoinTables.dwsOcpxWxAppOceanGenderAudienceReportData();
        sourceSqlList.add(genderSqls.get(sourceType));
        sinkSqlList.add(genderSqls.get(sinkType));
        insertSqlList.add(genderSqls.get(insertType));

        //抖音年龄画像
        Map<String, String> ageSqls = DwsLookUpJoinTables.dwsOcpxWxAppOceanAgeAudienceReportData();
        sourceSqlList.add(ageSqls.get(sourceType));
        sinkSqlList.add(ageSqls.get(sinkType));
        insertSqlList.add(ageSqls.get(insertType));

        //抖音城市画像
        Map<String, String> citySqls = DwsLookUpJoinTables.dwsOcpxWxAppOceanCityAudienceReportData();
        sourceSqlList.add(citySqls.get(sourceType));
        sinkSqlList.add(citySqls.get(sinkType));
        insertSqlList.add(citySqls.get(insertType));*/

/*        //千川消耗数据
        Map<String, String> adServingSqls = DwsLookUpJoinTables.dwsOcpxAdServingReport();
        sourceSqlList.add(adServingSqls.get(sourceType));
        sinkSqlList.add(adServingSqls.get(sinkType));
        insertSqlList.add(adServingSqls.get(insertType));

        //百度消耗数据
        Map<String, String> BaiduCreativeSqls = DwsLookUpJoinTables.dwsOcpxBaiduCreativeReport();
        sourceSqlList.add(BaiduCreativeSqls.get(sourceType));
        sinkSqlList.add(BaiduCreativeSqls.get(sinkType));
        insertSqlList.add(BaiduCreativeSqls.get(insertType));

        //年龄画像数据
        Map<String, String> ageDataSqls = DwsLookUpJoinTables.dwsOcpxPlatformAgeReportData();
        sourceSqlList.add(ageDataSqls.get(sourceType));
        sinkSqlList.add(ageDataSqls.get(sinkType));
        insertSqlList.add(ageDataSqls.get(insertType));

        //省份城市画像数据
        Map<String, String> cityDataSqls = DwsLookUpJoinTables.dwsOcpxPlatformCityReportData();
        sourceSqlList.add(cityDataSqls.get(sourceType));
        sinkSqlList.add(cityDataSqls.get(sinkType));
        insertSqlList.add(cityDataSqls.get(insertType));

        //性别画像数据
        Map<String, String> genderDataSqls = DwsLookUpJoinTables.dwsOcpxPlatformGenderReportData();
        sourceSqlList.add(genderDataSqls.get(sourceType));
        sinkSqlList.add(genderDataSqls.get(sinkType));
        insertSqlList.add(genderDataSqls.get(insertType));*/


        Map<String, String> orderSqls = DwdLookUpJoinTablesN.dwdPayOrder();
        sourceSqlList.add(orderSqls.get(sourceType));
        sinkSqlList.add(orderSqls.get(sinkType));
        insertSqlList.add(orderSqls.get(insertType));


        Map<String, String> wxuserSqls = DwdLookUpJoinTablesN.odsWxuser();
        sourceSqlList.add(wxuserSqls.get(sourceType));
        sinkSqlList.add(wxuserSqls.get(sinkType));
        insertSqlList.add(wxuserSqls.get(insertType));


        Map<String, String> accountSqls = DwdLookUpJoinTablesN.odsAccount();
        sourceSqlList.add(accountSqls.get(sourceType));
        sinkSqlList.add(accountSqls.get(sinkType));
        insertSqlList.add(accountSqls.get(insertType));


        res.put(sourceType,sourceSqlList);
        res.put(sinkType,sinkSqlList);
        res.put(insertType,insertSqlList);

        return res;
    }

}
